package cn.itcast.spark.sql

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.types.{DoubleType, IntegerType, LongType, StructField, StructType}
import org.junit.Test

class NullProcessor {

  val spark = SparkSession.builder()
    .master("local[6]")
    .appName("null")
    .getOrCreate()

  import spark.implicits._

  @Test
  def nullAndNaN(): Unit = {

    // 1. 读取数据集
    // 1.1 通过spark-csv自动的推断类型来读取，推断数字的时候会将NaN推断为字符串
//    spark.read.option("header", true).option("inferSchema", true).csv()

    // 1.2 直接读取字符串，在后续的操作中使用map算子转类型
//    spark.read.csv().map(row => row...)

    // 1.3 指定schema，不要自动推断
    val schema = StructType(
      List(
        StructField("id", LongType),
        StructField("year", IntegerType),
        StructField("month", IntegerType),
        StructField("day", IntegerType),
        StructField("hour", IntegerType),
        StructField("season", IntegerType),
        StructField("pm", DoubleType)
      )
    )

    val sourceDF = spark.read
      .option("header", value = true)
      .schema(schema)
      .csv("dataset/beijingpm_with_nan.csv")

    sourceDF.show()


    // 2. 丢弃
    // 2.1 any，只要有一个NaN就丢弃
    sourceDF.na.drop("any").show()
    sourceDF.na.drop().show()
    // 2.2 all，所有的列都是NaN才丢弃
    sourceDF.na.drop("all").show()
    // 2.3 某些列的规则
    sourceDF.na.drop("any", List("year", "month", "day", "hour")).show()


    // 3. 填充
    // 3.1 针对所有列数据进行默认值填充
    sourceDF.na.fill(0).show()

    // 3.2 针对特定列填充
    sourceDF.na.fill(0, List("year", "month")).show()

  }



}





















